package com.minelx.datamicroscope.core.condition;

import com.alibaba.fastjson.JSONObject;
import com.minelx.datamicroscope.core.query.AliasedKey;
import com.minelx.datamicroscope.core.values.CommonField;
import com.minelx.hellopandas.core.DataFrameDescription;
import io.gitee.minelx.commontools.factor.CloseableIterator;
import io.gitee.minelx.commontools.factor.Factor;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.function.Supplier;

import static java.util.stream.Collectors.toMap;

class RawInFromStreamSourceCondition implements RawCondition {

    private final List<AliasedKey> keys;

    private final Supplier<InputStream> streamSupplier;

    private final DataFrameDescription description;

    RawInFromStreamSourceCondition(List<AliasedKey> keys,
                                   Supplier<InputStream> streamSupplier,
                                   DataFrameDescription description) {
        this.keys = keys;
        this.streamSupplier = streamSupplier;
        this.description = description;
    }

    @Override
    public Factor<Condition> toFactor() {
        return factorFactory().factor();
    }

    private ConditionFactorFactory factorFactory() {
        if (keys.size() == 1) {
            CommonField onlyField = keys.get(0).getKey();
            return new SingleKeyStreamSourceFactorFactory(
                    onlyField,
                    () -> csvContentAsLineIterator(streamSupplier.get()));
        }
        // union keys
        return new UnionKeyStreamSourceFactorFactory(keys, streamSupplier);
    }

    private CloseableIterator<String> csvContentAsLineIterator(InputStream inputStream) {
        BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
        try {
            reader.readLine(); // consume the first line
        } catch (IOException e) {
            throw new RuntimeException("error while doing...", e);
        }
        return CloseableIterator.of(
                reader.lines().map(String::trim).iterator(),
                inputStream);
    }

    @Override
    public JSONObject toJSON() {
        JSONObject result = new JSONObject();
        result.put("operator", "in");
        result.put("file", description.dumpToJSON());
        result.put("nameMappings", keys.stream().collect(toMap(
                AliasedKey::getAlias,
                aliasedKey -> aliasedKey.getKey().name())));
        return result;
    }
}
